package com.ford.cn.csre.scenerio.domain.vo;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.drools.core.impl.KnowledgeBaseImpl;
import org.kie.api.KieBaseConfiguration;
import org.kie.api.KieServices;
import org.kie.api.conf.EventProcessingOption;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;
import org.kie.internal.builder.KnowledgeBuilder;
import org.kie.internal.builder.KnowledgeBuilderFactory;
import org.kie.internal.io.ResourceFactory;
import org.kie.internal.utils.KieHelper;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class RuleAlarmFunction extends RichCoFlatMapFunction<Alarm, Rule, Alarm> {
    private  KieHelper kieHelper;
    private  KnowledgeBaseImpl kieBase;
    private  KieSession kieSession;
    private  EntryPoint entryPoint;
    private Lock lock = new ReentrantLock();

    @Override
    public void open(Configuration parameters) throws Exception {
        if(kieSession == null ){
            kieHelper = new KieHelper();
            //kieHelper.addContent(drlStr, ResourceType.DRL);
            KieBaseConfiguration config = KieServices.Factory.get().newKieBaseConfiguration();
            config.setOption( EventProcessingOption.STREAM );
            try{
                kieBase =(KnowledgeBaseImpl) kieHelper.build();
                kieSession = kieBase.newKieSession();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    @Override
    public void close() throws Exception {
        Thread.sleep(1000);
        kieSession.destroy();
    }


    @Override
    public void flatMap1(Alarm alarm, Collector collector) throws Exception {
        lock.lock();
        try{
            entryPoint = kieSession.getEntryPoint("Demo02");
            if(entryPoint != null){
                entryPoint.insert(alarm);
                int fireNum = kieSession.fireAllRules();
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        collector.collect(alarm);
    }

    @Override
    public void flatMap2(Rule rule, Collector collector) throws Exception {
        if(rule.getOperate() == 0){
            lock.lock();
            try{
                kieBase.removeRule("rules", rule.getName());
                kieSession = kieBase.newKieSession();
            }finally {
                lock.unlock();
            }
        }else{
            // 规则名是ruleId
            org.kie.api.definition.rule.Rule ruleCache = kieBase.getRule("rules", rule.getName());

            // 规则如果已存在，不需要再次新增
            if(ruleCache !=null && rule.getName().equals(ruleCache.getName())){
                //System.out.println("规则已经存在，ruleName="+ ruleCache.getName());
            }else {
                //重新添加规则
                KnowledgeBuilder kb = KnowledgeBuilderFactory.newKnowledgeBuilder();
                //装入规则，可以装入多个
                kb.add(ResourceFactory.newByteArrayResource(rule.getDrlStr().getBytes("utf-8")), ResourceType.DRL);
                kieBase.addPackages(kb.getKnowledgePackages());
                kieSession = kieBase.newKieSession();
            }
        }
    }
}